package com.larry.kafka.producer;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;

/**
 * @ProjectName hadoop
 * @Author larry
 * @Date 20:31 2021/12/1
 * @Description 生产者 异步发送 回调
 **/
public class KafkaProducerDemo1 {

    public static void main(String[] args) {
        //创建配置对象
        Properties properties = new Properties();
        properties.put("bootstrap.servers","192.168.10.102:9092");

        //ack级别
        properties.put("acks","all");
        //重试次数
        properties.put("retries",1);
        //批次大小
        properties.put("batch.size",16384);
        //等待时间
        properties.put("linger.ms",1);

        properties.put("buffer.memory",33554432);

        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        //创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);


        for (int i = 0; i < 10; i++) {
            //producer.send(new ProducerRecord<String,String>("first","a" + i));
            producer.send(new ProducerRecord<String, String>("first", 0, null, "a" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    String topic = metadata.topic();
                    System.out.println(topic);
                    if (exception != null){
                        System.out.println("消息发送失败: " + exception.getMessage());
                    }
                }
            });
        }

        producer.close();

    }

}
